package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SparkSession}

object Code08DSL {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("dsl")
      .config("spark.sql.shuffle.partitions", "3")
      .getOrCreate()

    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      // 1500100001,施笑槐,22,女,文科六班
      .schema("id String,name String,age int,gender String,clazz String")
      .load("spark_code/data/students.txt")


    /**
     * SQL的处理逻辑：
     *
     * 关键字：
     *  FROM -> JOIN -> ON -> WHERE -> GROUP BY -> HAVING -> SELECT -> ORDER BY  -> LIMIT
     *
     * SQL函数
     *    字符串函数 日期函数 数值函数 聚合函数  窗口函数 ...
     *
     *
     * DSL:
     * FROM 可以在Spark中进行做read数据读取
     * WHERE 在DSL语法中存在有三种写法 具体在后续代码中查看
     * GROUP BY 可以给定 $"列名" "列名" 同时对于 $"列名" 可以添加相对应的函数 以及表达式
     * 由于GROUP BY 过程中会产生shuffle shuffle过程中会使用默认并发数 200 所以可以使用
     * .config("spark.sql.shuffle.partitions","3") 来设置shuffle过程中的并发量
     * HAVING  在DSL语言中并不存在，可以使用WHERE语句对其进行过滤， 由此可以知道DSL与SQL相比，并不需要遵守SQL的语法规则
     * SELECT  可以在SELECT中选取列，以及给定表达式，或者调用函数 但是和SQL中不一样的是 GROUP BY 后不能使用select
     * JOIN   可以将两个DF进行关联，关联时，默认为inner join
     *        关联条件给定方式：
     *            1.如果列名称相同，那么可以直接给定一个 “列名”
     *            2.如果有多个列，那么可以给定一个 List("列名")
     *            3.如果给定条件表达式，那么可以使用 $"列名" 运算符  $"列名" => 依据返回的是 Column对象
     *            4.如果给定条件表达式，还可以使用 expr("SQL表达式")  => 依据返回的是 Column对象
     *            5.如果给定条件表达式，还可以通过 df("列名") 运算符 df("列名") 该方式可以指定从哪些DF中取出
     *
     *            注意：如果存在多个条件，可以使用 and 符号进行连接
     *
     *  ORDER BY  对指定的列进行排序，其表达式可以给定列名 $列名 默认为升序 可以在 $列名后添加 desc 表示倒叙
     *            注意：多个列可以直接添加多个列名称
     *
     *  LIMIT  对数据进行限制条数  Spark底层会对其进行优化，执行效率更高
     *
     */

    // sql函数
    import org.apache.spark.sql.functions._
    // Spark隐式转换
    import spark.implicits._


    stuDF
      // 在where中如果直接传入一个 字符串并不能表示一个表达式
      //  在DSL中需要使用 === 号来进行做等值判断 ===的方法是 ColumnName 中所存在的方法
      //  而 ColumnName 表示为一个列,其获取方式为 $"列名" 进行获取
      //  而 $ 又是 spark 隐式转换中的一个方法,而 spark为当前SparkSession的一个对象
      //      .where($"clazz" === "文科一班")
      // 该方法是在where条件中给定一个SQL表达式，并且expr是functions中的一个方法
      //  传入的字符串为SQL的表达式
      .where(expr("clazz = '文科一班'"))
      // groupBy中可以指定分组的列 而列可以使用字符串 同时也可以使用 $"列名称"
      //      .groupBy("gender","clazz")
      // $"列名称"返回的为ColumnName对象属于Column子类 ColumnName(name: String) extends Column(name)
      .groupBy($"gender", $"clazz")
      // count是调用 functions._ 中的函数 agg中可以传入聚合函数 包括 max min count sum avg 等..
      .agg(count("*"))
      .show()


    stuDF
      //      .where(expr("age > 21"))
      //      .where($"age" > 21)
      // 传入一个字符串的形式
      .where("age > 21")
      .groupBy($"clazz")
      .agg(count("*") as "cnt")
      .where($"cnt" > 30)
      .show()


    // 统计理科和文科班各有多少人

    stuDF
      //      .groupBy(substring($"clazz",0,2) as "sub_clazz")
      .groupBy(expr("substring(clazz,0,2)") as "sub_clazz")
      //      .agg(count("*") as "cnt")
      .count()

      .show()


    //    stuDF
    ////      .select($"id",$"age" as "new_age",$"gender")
    ////      .select(expr("age + 1" ) as "new_age",substring($"id",7,4))
    //      .select(expr("age + 1" ) as "new_age",expr("substring(id,7)") as "sub_id")
    //      .show()


    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      // 1500100001,施笑槐,22,女,文科六班
      .schema("id String,courseID String,score int")
      .load("spark_code/data/score.txt")

    scoreDF
      .groupBy("id")
      .agg(sum($"score") as "total_score")
      .orderBy($"total_score".desc)
      .show()


    scoreDF
      .groupBy("id")
      .agg(sum($"score") as "total_score")
      .join(stuDF, List("id"))
      .orderBy($"clazz", $"total_score".desc)
      .show()


    //    println(countNum)
    //      .show()

    scoreDF
      .groupBy("id")
      .agg(sum($"score") as "total_score")
      .join(stuDF,"id") // 使用相同列ID进行关联
      .show()

  }
}
